package com.atguigu.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.atguigu.gmall.realtime.app.func.DimSink;
import com.atguigu.gmall.realtime.app.func.TableProcessFunction;
import com.atguigu.gmall.realtime.beans.MyDeserializationSchemaFunction;
import com.atguigu.gmall.realtime.beans.TableProcess;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;

/**
 * Author: Felix
 * Date: 2021/9/15
 * Desc: 业务数据的动态分流
 */
public class BaseDBApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);

        /*//TODO 2.设置检查点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000L));
        env.setStateBackend(new FsStateBackend("xxx"));
        System.setProperty("HADOOP_USER_NAME","atguigu");
        */
        //TODO 3.从Kafka读取业务数据
        //3.1 声明消费的主题以及消费者组
        String topic = "ods_base_db_m";
        String groupId = "base_db_app_groupId";

        //3.2 获取消费者对象
        FlinkKafkaConsumer<String> kafkaSourceFunction = MyKafkaUtil.getKafkaSourceFunction(topic, groupId);

        //3.3 读取数据  封装为流
        DataStreamSource<String> kafkaDS = env.addSource(kafkaSourceFunction);

        //TODO 4.对读取的数据类型进行转换   jsonStr--->jsonObj
        /*
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map(
            new MapFunction<String, JSONObject>() {
                @Override
                public JSONObject map(String jsonStr) throws Exception {
                    return JSON.parseObject(jsonStr);
                }
            }
        );
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map(
            jsonStr -> JSON.parseObject(jsonStr)
        );
        */
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map(JSON::parseObject);

        //TODO 5.做一个简单的etl
        SingleOutputStreamOperator<JSONObject> filterDS = jsonObjDS.filter(
            new FilterFunction<JSONObject>() {
                @Override
                public boolean filter(JSONObject jsonObj) throws Exception {
                    boolean flag = jsonObj.getString("table") != null
                        && jsonObj.getString("table").length() > 0
                        && jsonObj.getJSONObject("data") != null
                        && jsonObj.getString("data").length() > 3;
                    return flag;
                }
            }
        );

        //filterDS.print(">>>>");

        //TODO 6.使用FlinkCDC读取配置信息
        //6.1 通过FlinkCDC获取MySQLSourceFunction
        DebeziumSourceFunction<String> mySQLSourceFunction = MySQLSource.<String>builder()
            .hostname("hadoop202")
            .port(3306)
            .username("root")
            .password("123456")
            .databaseList("gmall0408_realtime")
            .tableList("gmall0408_realtime.table_process")
            .deserializer(new MyDeserializationSchemaFunction())
            .startupOptions(StartupOptions.initial())
            .build();
        //6.2 从mysql中读取配置表数据  封装为流
        DataStreamSource<String> mySQLDS = env.addSource(mySQLSourceFunction);

        //TODO 7.将配置流转换为广播流，并指定广播状态描述器
        //7.1 创建广播状态描述器
        MapStateDescriptor<String, TableProcess> mapStateDes = new MapStateDescriptor<>("tableProcess", String.class, TableProcess.class);

        //7.2 将普通配置流转换为广播流
        BroadcastStream<String> broadcastDS = mySQLDS.broadcast(mapStateDes);

        //TODO 8.将业务流和配置流进行连接
        BroadcastConnectedStream<JSONObject, String> connectDS = filterDS.connect(broadcastDS);

        //TODO 9.对连接之后的流进行处理----分流  （维度---维度侧输出流   事实---主流）
        OutputTag<JSONObject> dimTag = new OutputTag<JSONObject>("dimTag") {};

        SingleOutputStreamOperator<JSONObject> realDS = connectDS.process(new TableProcessFunction(dimTag, mapStateDes));

        DataStream<JSONObject> dimDS = realDS.getSideOutput(dimTag);

        realDS.print(">>>");
        dimDS.print("###");

        //TODO 10.将维度数据写到Hbase
        dimDS.addSink(new DimSink());

        //TODO 11.将主流数据写到Kafka
        realDS.addSink(MyKafkaUtil.getKafkaSinkFunctionBySchema(new KafkaSerializationSchema<JSONObject>() {
            @Override
            public ProducerRecord<byte[], byte[]> serialize(JSONObject jsonObj, @Nullable Long timestamp) {
                String topic = jsonObj.getString("sink_table");
                return new ProducerRecord<byte[], byte[]>(topic,jsonObj.getJSONObject("data").toJSONString().getBytes());
            }
        }));

        env.execute();
    }
}
